Add direct consumer support #169

sircinek · 2020-06-21T16:55:32Z

This PR aims at adding support for direct consumer.
It provides default callback module implementing amqp_gen_consumer behavior, based on erlang client library amqp_direct_consumer.

This can be useful when you want to maintain 1 to 1 relationship, between channel and consumer.
It also puts your consumer directly under amqp tree, making it a little less error/race condition prone, as well as simplifying message flow between channel and user defined consumer (1 layer less).

PR consists of some hygiene changes, updating deprecated functions and addressing dialyzer/credo issues as well as bumping up dependencies versions.

…ntation.

…dules and init args for direct consumer mode.

ono · 2020-06-22T07:52:18Z

Hi @sircinek. Thanks for PR. I need to give a proper look later but it looks to be a good addition. Before that...

PR consists of some hygiene changes, updating deprecated functions and addressing dialyzer/credo issues as well as bumping up dependencies versions.

Could you remove these changes from this PR? These should be discussed in separate PR(s) and I would like you to include only amqp_direct_consumer changes in this PR 🙏

sircinek · 2020-06-22T20:32:01Z

Hey @ono !

Could you remove these changes from this PR? These should be discussed in separate PR(s) and I would like you to include only amqp_direct_consumer changes in this PR 🙏

Absolutely, reverted all unrelated changes. Do you want me to open a separate PR, or you want to take them in smaller chunks ?

ono · 2020-06-27T11:16:10Z

Great. Let's get this PR done at first. I am hoping to make some time for this PR soon. Bear with me.

sircinek · 2020-06-29T11:11:55Z

lib/amqp/channel.ex


  @doc """
  Opens a new Channel in a previously opened Connection.
+
+  ## Options
+    * `:consumer_type` - specifies the type of consumer used as callback module.


Maybe this should be :default or :custom as currently :direct option lets you pass anything there.
It could also be more restrictive and allow only default callback module (DirectReceiver) implementation - which was my original intention - then this control flag had more sense. @ono what do you think?

:default and :direct might be my preference. I am not comfortable with :custom sets DirectReceiver as a default consumer. :direct makes more sense to me. we can introduce :custom in future if required as an additional option.

The other option is to eliminate this option and take {consumer_module, args} as a tuple like Erlang library does. Then make the following changes:

Rename DirectReciver to DirectConsumer

Add a proper documentation to DirectConsumer module

Then the user can use it simply like this:

chan = AMQP.Channel.open(conn) # call :amqp_connection.open_channel/1 chan = AMQP.Channel.open(conn, {AMQP.DirectConsumer, self()}) # call :amqp_connection.open_channel/2

This approach makes more sense to me as direct or non-direct is not a right concern on channel layer. What do you think?

Will do the latter, as it is better aligned with how Erlang library works.
Side note - maybe we would like to change Receiver to implement amqp_gen_consumer and be a callback module substituting default amqp_selective_receiver as it more or less does the same thing ? As a separate PR ofc.

The main purpose of Receiver is...

to convert the data from Erlang library(Record) to Elixir friendly data type(Map)

to receive messages for a channel in a single process so we can ensure the order of messages (deliver, returns etc.)

Now... I am a bit lost of the purpose of this PR 😓

To align 100%, could you write down an example code with direct_consumer and how it solves the problems?

haljin · 2020-07-20T08:54:04Z

Hi, is there any progress in this? We could also use this functionality.

ono · 2020-07-20T13:51:55Z

Hi, is there any progress in this? We could also use this functionality.

Yes, sorry I have been bit too busy with other stuff but should be able to look into it in this week. Thanks for your patience 🙏

ono

I am really sorry about my slow response. I have been occupied with other business.

Great work 👍 I am keen on adding this to the project. Require a few changes so please check out my comments in code.

Thanks!

ono · 2020-07-23T20:12:37Z

lib/amqp/channel.ex


  @doc """
  Opens a new Channel in a previously opened Connection.
+
+  ## Options
+    * `:consumer_type` - specifies the type of consumer used as callback module.


:default and :direct might be my preference. I am not comfortable with :custom sets DirectReceiver as a default consumer. :direct makes more sense to me. we can introduce :custom in future if required as an additional option.

ono · 2020-07-23T20:22:03Z

lib/amqp/basic.ex


-    case :amqp_channel.subscribe(chan.pid, basic_consume, receiver.pid) do
+        :direct ->
+          consumer_pid


Do we still need to subscribe direct consumer pid? Isn't it subscribed automatically when channel is opened?

AFAIU, amqp_channel.subscribe/3 starts a consumer for a queue that is bound to specific channel (and shares its lifetime (or until cancelled)), with opts specified in #basic_consume{}. The pid is in fact ignored by our (and rabbitmq) implementation of direct consumer, but the server method needs to be sent anyway.
Basically we can pass any pid there, but it doesn't make sense to start extra receiver as it will not get any messages.

Why do you want to subscribe a channel when you have a direct consumer for it?

ono · 2020-07-23T20:25:45Z

test/basic_test.exs

@@ -19,100 +19,124 @@ defmodule BasicTest do
    assert :ok = Basic.publish(meta[:chan], "", "", "ping")
  end

-  test "basic return", meta do
-    :ok = Basic.return(meta[:chan], self())
+  for cons <- [:selective, :direct] do


It is making test's readability lower. Can we leave the existing tests as they are? Then add couple tests which test direct consumer behaviours? It can be a separate test file like direct_consumer_test.exs.

Just to clarify - I am not suggesting copy paste all tests to direct_consumer_test.exs. Probably a good idea to write a test for direct consumer which covers open channel, send and consume a message, close a channel etc.

Sure, will do. I just didn't want to duplicate code, as existing tests could cover both cases. I do agree though on readability argument.

haljin · 2020-09-21T12:52:45Z

What's the state of this PR? I'd gladly take it over as we still could really use that change. @sircinek ?

ono · 2020-10-03T08:23:36Z

this work is handed over to #172

sircinek added 2 commits June 14, 2020 20:46

Added default callback module and support for direct consumer impleme…

f266f3b

…ntation.

Update documentation. Expand options to pass user defined callback mo…

d2e9126

…dules and init args for direct consumer mode.

Revert unrelated changes.

6a14f31

sircinek commented Jun 29, 2020

View reviewed changes

ono requested changes Jul 23, 2020

View reviewed changes

haljin mentioned this pull request Oct 3, 2020

Add direct consumer support #172

Merged

ono closed this Oct 3, 2020

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add direct consumer support #169

Add direct consumer support #169

sircinek commented Jun 21, 2020 •

edited

Loading

ono commented Jun 22, 2020

sircinek commented Jun 22, 2020

ono commented Jun 27, 2020

sircinek Jun 29, 2020

ono Jul 23, 2020

ono Jul 24, 2020

sircinek Jul 27, 2020 •

edited

Loading

ono Jul 27, 2020

haljin commented Jul 20, 2020

ono commented Jul 20, 2020

ono left a comment

ono Jul 23, 2020

ono Jul 23, 2020

sircinek Jul 27, 2020 •

edited

Loading

ono Jul 27, 2020

ono Jul 23, 2020

ono Jul 23, 2020

sircinek Jul 27, 2020

haljin commented Sep 21, 2020

ono commented Oct 3, 2020

Add direct consumer support #169

Add direct consumer support #169

Conversation

sircinek commented Jun 21, 2020 • edited Loading

ono commented Jun 22, 2020

sircinek commented Jun 22, 2020

ono commented Jun 27, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sircinek Jul 27, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

haljin commented Jul 20, 2020

ono commented Jul 20, 2020

ono left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sircinek Jul 27, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

haljin commented Sep 21, 2020

ono commented Oct 3, 2020

sircinek commented Jun 21, 2020 •

edited

Loading

sircinek Jul 27, 2020 •

edited

Loading

sircinek Jul 27, 2020 •

edited

Loading